package com.jie.flink.cdc.util;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Iterator;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author zhanggj
 * @date 2023/3/6 18:29
 * @desc
 */
@Slf4j
public class LogPrintLimitUtil {


    /**
     * 日志打印间隔条数
     */
    private static final int LOG_OUT_MAX_TIMES = 256;

    /**
     * 日志打印最小时间间隔
     */
    private static final long LOG_OUT_MIN_INTERVAL_TIME = 5 * 1000L;

    /**
     * 日志打印最大时间间隔
     */
    private static final long LOG_OUT_MAX_INTERVAL_TIME = 30 * 1000L;

    /**
     * 日志打印时间
     */
    private static final ConcurrentHashMap<String, Long> LOG_START_MAP = new ConcurrentHashMap<>(512);

    /**
     * 日志数量
     */
    private static final ConcurrentHashMap<String, Long> LOG_NUM_MAP = new ConcurrentHashMap<>(512);

    /**
     * 日志打印对象
     */
    private static final ConcurrentHashMap<String, Logger> LOG_MAP = new ConcurrentHashMap<>(128);

    private static ScheduledExecutorService scheduledExecutorService;

    private static final String MAP_KEY_SEPARATOR = "_&@&_";

    private static final String LOG_FORMAT = "从{}到现在，共需要打印{}次：{}";

    private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    public static void printDelayLog(Logger log, String message) {
        // 开启打印进程
        startPrintThread();

        final String key = log.getName().concat(MAP_KEY_SEPARATOR).concat(message);
        LOG_MAP.put(key, log);

        synchronized (LOG_NUM_MAP) {
            Long logStart = LOG_START_MAP.putIfAbsent(key, System.currentTimeMillis());
            if (Objects.isNull(logStart)) {
                LOG_NUM_MAP.put(key, 1L);
            } else {
                LOG_NUM_MAP.put(key, LOG_NUM_MAP.get(key) + 1);
            }
        }
    }

    private static void startPrintThread() {
        if (Objects.nonNull(scheduledExecutorService)) {
            return;
        }
        synchronized (LOG_MAP) {
            if (Objects.nonNull(scheduledExecutorService)) {
                return;
            }
            scheduledExecutorService = new ScheduledThreadPoolExecutor(1);
        }
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            final Iterator<String> keyIterator = LOG_START_MAP.keySet().iterator();
            final Long now = System.currentTimeMillis();
            while(keyIterator.hasNext()) {
                final String key = keyIterator.next();
                // 数量多或者时间长的打印
                if (LOG_NUM_MAP.get(key) >= LOG_OUT_MAX_TIMES || LOG_START_MAP.get(key) >= now - LOG_OUT_MAX_INTERVAL_TIME) {
                    // 打印日志
                    LOG_MAP.get(key).info(LOG_FORMAT, sdf.format(new Date(LOG_START_MAP.get(key))), LOG_NUM_MAP.get(key), StringUtils.substringAfter(key, MAP_KEY_SEPARATOR)
                          );
                    synchronized (LOG_NUM_MAP) {
                        LOG_NUM_MAP.remove(key);
                        LOG_START_MAP.remove(key);
                    }
                }
            }
        }, LOG_OUT_MIN_INTERVAL_TIME, LOG_OUT_MIN_INTERVAL_TIME, TimeUnit.MILLISECONDS);
    }

}
